package com.hudsonmq.spring.client;

import com.hudson.codec.MessageCodecFactory;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudsonmq.spring.config.HudsonMQConfig;
import com.hudsonmq.spring.handler.netty.ProcesserHandler;
import com.hudsonmq.spring.handler.netty.ReconnectionHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/6 23:08
 **/
@Component
public class BrokerClient implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(BrokerClient.class);

    @Resource
    private HudsonMQConfig hudsonMQConfig;

    private static final NioEventLoopGroup group = new NioEventLoopGroup(50);
    private static Bootstrap bootstrap = null;

    public static ChannelFuture getChannel(String brokerName, String address, int port) {
        ChannelFuture channelFuture = bootstrap.connect(address, port);
        channelFuture.addListener((ChannelFutureListener) f -> {
            if (f.isSuccess()) {
                logger.info("连接建立成功");
                f.channel().pipeline().addLast(new ReconnectionHandler(brokerName, address, port));
                logger.debug("成功连接上 Broker 服务器： {}", address + ":" + port);
            } else {
                logger.error("连接失败: {}", f.cause().getMessage());
            }
        });
        return channelFuture;
    }

    public static void getTopicQueueInfo(Channel channel) {
        channel.writeAndFlush(
                RemotingCommand.builder()
                        .type(MessageTypeEnum.QUEUE_MAP.getType())
                        .build());
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        bootstrap = new Bootstrap().group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<NioSocketChannel>() {
            @Override
            protected void initChannel(NioSocketChannel sc) throws Exception {
                sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
                sc.pipeline().addLast(MessageCodecFactory.getCodec(hudsonMQConfig.getSerializableMethod()));
                sc.pipeline().addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS)); // 写空闲4秒触发 WRITE_IDLE
                sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                    @Override
                    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                        if (evt instanceof IdleStateEvent) {
                            IdleStateEvent event = (IdleStateEvent) evt;
                            if (event.state() == IdleState.WRITER_IDLE) {
                                // 发送心跳包
                                logger.debug("produce 空闲，发送心跳至 broker...");
                                ctx.writeAndFlush(RemotingCommand.builder().type(MessageTypeEnum.IDLE_MESSAGE.getType()).build()); // 或者发送自定义心跳消息
//                                        new Message<>(MessageTypeEnum.IDLE_MESSAGE.getType(), null));
                            }
                        }
                    }
                });
                sc.pipeline().addLast(new ProcesserHandler());
            }
        });
    }
}

